package concurrency

import (
	"cow_golang/shared_mod/logger"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

// 支持不同策略的并发控制
// 主要目的让业务实现的逻辑层减少开发并发的逻辑，让业务逻辑层开发
//启动Worker
//↓
//等待queueChan或stopChan
///       \
//收到队列      收到停止信号
//↓               ↓
//CAS抢占处理权     结束协程
//↓
//进入处理循环
//↓
//取任务 → 执行 → 继续循环
//↓
//无任务时退出循环
//↓
//检查剩余任务 → 有则重新入队
//↓
//回到主循环

type Queue struct {
	// 任务队列
	taskChan chan Job
	// 正在被处理 标识
	processing int32
}

type MultWorkQueue struct {
	// 不同的任务队列(key 玩家id ， 模块名称， 场景id) value 对应的任务队列0
	typeQueues sync.Map
	// 有任务的工作队列，加入到channel，goroutine获取，后进行处理
	queueChan chan *Queue
	// 工作goroutine数量
	maxWorkers int
	// 停止工作chan
	stopChan chan struct{}
	wg       sync.WaitGroup
}

// 获取并发模块
func NewMultWorkQueue(maxWorkers int) *MultWorkQueue {
	p := &MultWorkQueue{
		typeQueues: sync.Map{},
		queueChan:  make(chan *Queue, 1024),
		maxWorkers: maxWorkers,
		stopChan:   make(chan struct{}),
	}
	// 启动工作goroutine
	p.wg.Add(maxWorkers)
	for i := 0; i < maxWorkers; i++ {
		fmt.Printf("concurrency start worker %d\n", i)
		go p.worker()
	}

	return p
}

// 停止并发处理mod
func (p *MultWorkQueue) Stop() {
	// 关闭停止消息通道
	close(p.stopChan)
	// 等待worker goroutine退出
	p.wg.Wait()
	p.typeQueues.Range(func(key, value interface{}) bool {
		// 任务加入到任务队列
		taskChan, ok := value.(Queue)
		if ok {
			close(taskChan.taskChan)
		}
		return true
	})
}

// 加入处理任务
func (p *MultWorkQueue) AddTask(key string, job Job) {
	// 加锁，保护map
	//获取自己key 对应的任务队列
	q, _ := p.typeQueues.LoadOrStore(key, Queue{
		taskChan:   make(chan Job, 1024),
		processing: 0,
	})
	// 任务加入到任务队列
	taskChan := q.(Queue)
	taskChan.taskChan <- job
	p.tryEnqueueQueue(&taskChan)
}

// 增加带有回复的安全执行函数
func (p *MultWorkQueue) safeExec(job Job) {
	start := time.Now()
	defer func() {
		// 日志记录下处理时间
		logger.GetInstance().Warn("Execute elapse:", time.Since(start))
	}()
	// 出错回复处理
	defer func() {
		if r := recover(); r != nil {
			logger.GetInstance().Warnf("捕获到panic: %v\n", r) // 可替换为日志记录
		}
	}()
	// 执行实际任务
	job.Execute()
}

// 具体的工作goroutine，循环抢占 处理标识
func (p *MultWorkQueue) worker() {
	defer p.wg.Done()

	for {
		select {
		//有任务需要处理
		case q := <-p.queueChan:
			fmt.Printf("concurrency worker have task, q %d\n", q.processing)
			if atomic.CompareAndSwapInt32(&q.processing, 0, 1) {
				fmt.Printf("concurrency worker grab  task\n")
				hasMore := true
				for hasMore {
					select {
					case task := <-q.taskChan:
						p.safeExec(task)
					default:
						// 队列暂时为空
						atomic.StoreInt32(&q.processing, 0)
						// 检查是否有残留任务需要重新入队
						if len(q.taskChan) > 0 {
							p.tryEnqueueQueue(q)
						}
						hasMore = false // 退出内层循环
					}
				}
			}
		case <-p.stopChan:
			return
		}
	}
}

// 有任务的队列，加入到处理channel
func (p *MultWorkQueue) tryEnqueueQueue(q *Queue) {
	if atomic.CompareAndSwapInt32(&q.processing, 0, 1) {
		p.queueChan <- q
		atomic.StoreInt32(&q.processing, 0)
	}
}
